package com.ccb.posrecord;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

/**
 * 通过重写LineRecordReader.nextKeyValue()，在LineRecordReader.nextKeyValue()的结果上实现： 将多条记录处理后做为一个完整的记录返回给mapper处理
 * 
 * @author flytoice
 *
 */
public class PosCustomRecordReader extends LineRecordReader {

	private LongWritable key;
	private Text value;

	@SuppressWarnings("rawtypes")
	Class CustomRecordClass;

	public PosCustomRecordReader(@SuppressWarnings("rawtypes") Class customRecord, byte[] recordDelimiter) {
		super(recordDelimiter);
		this.CustomRecordClass = customRecord;
	}

	@Override
	public boolean nextKeyValue() throws IOException {
		PosCustomRecord o = null;
		try {
			o = (PosCustomRecord) CustomRecordClass.newInstance();
		} catch (InstantiationException e) {
			throw new RuntimeException("InstantiationException");
		} catch (IllegalAccessException e) {
			throw new RuntimeException("IllegalAccessException");
		}

		if (super.nextKeyValue()) {
			while (true) {
				if (o.isYouNeedValue(super.getCurrentKey(), super.getCurrentValue(), o)) {
					key = super.getCurrentKey();
					value = o.getValueForMapper();
					return Boolean.TRUE;
				}
				if (!super.nextKeyValue()) {
					return Boolean.FALSE;
				}
			}
		} else {
			return Boolean.FALSE;
		}
	}

	@Override
	public LongWritable getCurrentKey() {
		return key;
	}

	@Override
	public Text getCurrentValue() {
		return value;
	}

}
